package com.boot.cluster.listener;

import com.boot.cluster.ClusterRaftClient;
import com.boot.cluster.model.MessageInfo;
import com.boot.cluster.utils.HessianSerializer;
import io.micrometer.core.instrument.util.NamedThreadFactory;
import org.jgroups.Message;
import org.jgroups.Receiver;
import org.jgroups.View;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class DispatchMessageListener implements Receiver {

	private static Logger logger = LoggerFactory.getLogger(DispatchMessageListener.class);
	private ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 60, TimeUnit.SECONDS,
			new LinkedBlockingQueue<>(),
			new NamedThreadFactory("dispatch-message-listener-thread-pool"));
	private ClusterRaftClient client;

	public DispatchMessageListener(ClusterRaftClient client) {
		this.client = client;
	}

	@Override
	public void receive(Message msg) {
		if (msg == null) {
			return;
		}
		try {
			MessageInfo<?> mi = HessianSerializer.deserialize(msg.getObject());
			if (mi != null) {
				mi.setSrc(msg.getSrc());
			}
			this.handle(mi, msg);
		} catch (Exception e) {
			logger.error("",e);
		}
	}


	private void handle(MessageInfo<?> mi,Message msg) {
		List<MessageReceiverListener> list = this.client.getMessageReceiverListener();
		if((list==null) || list.isEmpty()) {
			return;
		}
		if(mi.getType()==null) {
			list.forEach((s)->{
				try {
					this.executor.execute(()->s.receive(mi));
				} catch (Exception e) {
					logger.error("",e);
				}
			});
		}else {
			list.stream().filter(t -> t.getType().equals(mi.getType()))
					.forEach(s -> this.executor.execute(() -> {
						s.receive(mi);
					}));
		}
	}

	@Override
	public void viewAccepted(View view) {
		logger.info("Cluster:{},viewId:{}", view.getMembers(), view.getViewId());
		if (this.client.getGroupMemberListener() != null) {
			this.client.getGroupMemberListener().getMembers(
					view.getMembers());
		}
	}



}
